package com.spx.chapter05.transform;

import com.spx.chapter05.pojo.Event;
import com.spx.util.ClickSource;
import com.spx.util.SampleDataUtil;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.runtime.state.storage.FileSystemCheckpointStorage;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * create by undeRdoG on  2022-05-01  12:21
 * 凡心所向，素履以往，生如逆旅，一苇以航。
 */
public class ReduceTest {

    /**
    *   需求：返回访问量最大的用户
    * */
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        env.enableCheckpointing(6000L);

        env.setStateBackend(new HashMapStateBackend());

        CheckpointConfig checkpointConfig = env.getCheckpointConfig();

        checkpointConfig.setCheckpointStorage(new FileSystemCheckpointStorage("hdfs://hadoop102:8020/flink/checkpoints"));

        System.setProperty("HADOOP_USER_NAME", "shipeixin");

        DataStreamSource<Event> dataSource = env.addSource(new ClickSource());

        KeyedStream<Tuple2<String, Long>, String> keyedStream = dataSource.map(new MapFunction<Event, Tuple2<String, Long>>() {
            @Override
            public Tuple2<String, Long> map(Event value) throws Exception {
                return Tuple2.of(value.user, 1L);
            }
        }).keyBy(data -> data.f0);

        SingleOutputStreamOperator<Tuple2<String, Long>> userClicks = keyedStream.reduce(new ReduceFunction<Tuple2<String, Long>>() {
            @Override
            public Tuple2<String, Long> reduce(Tuple2<String, Long> value1, Tuple2<String, Long> value2) throws Exception {
                return Tuple2.of(value1.f0, value1.f1 + value2.f1);
            }
        });

        // 根据 userClicks统计出现频率最高的记录,此处为了上所有数据都到一个分组内进行统计，显式的keyBy(data -> "key") 使所有数据进到一个分组内
        SingleOutputStreamOperator<Tuple2<String, Long>> maxClick = userClicks.keyBy(data -> "key").reduce(new ReduceFunction<Tuple2<String, Long>>() {
            @Override
            public Tuple2<String, Long> reduce(Tuple2<String, Long> value1, Tuple2<String, Long> value2) throws Exception {
                if (value2.f1 > value1.f1) {
                    return Tuple2.of(value2.f0, value2.f1);
                } else {
                    return Tuple2.of(value1.f0, value1.f1);
                }
            }
        });

        SingleOutputStreamOperator<String> result = maxClick.map(new MapFunction<Tuple2<String, Long>, String>() {
            @Override
            public String map(Tuple2<String, Long> value) throws Exception {
                return "max click user is " + value.f0 + "， and number is " + value.f1;
            }
        });


        result.print();

        env.execute();


    }
}
